package com.wugui.datatx.core.thread;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.ArrayUtil;
import com.wugui.datatx.core.biz.AdminBiz;
import com.wugui.datatx.core.biz.model.HandleProcessCallbackParam;
import com.wugui.datatx.core.biz.model.ReturnT;
import com.wugui.datatx.core.enums.RegistryConfig;
import com.wugui.datatx.core.executor.JobExecutor;
import com.wugui.datatx.core.log.JobFileAppender;
import com.wugui.datatx.core.log.JobLogger;
import com.wugui.datatx.core.util.FileUtil;
import java.io.File;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

/** Created by jingwk on 2019/12/14. */
@Slf4j
public class ProcessCallbackThread {
  @Getter private static final ProcessCallbackThread instance = new ProcessCallbackThread();
  private static final String failCallbackFilePath =
      JobFileAppender.getLogPath()
          .concat(File.separator)
          .concat("processcallbacklog")
          .concat(File.separator);
  private static final String failCallbackFileName =
      failCallbackFilePath.concat("datax-web-processcallback-{x}").concat(".log");

  /** job results callback queue */
  private final LinkedBlockingQueue<HandleProcessCallbackParam> callBackQueue =
      new LinkedBlockingQueue<>();

  /** callback thread */
  private Thread processCallbackThread;

  private Thread processRetryCallbackThread;
  private volatile boolean toStop = false;

  public static void pushCallBack(HandleProcessCallbackParam callback) {
    getInstance().callBackQueue.add(callback);
    log.debug(
        ">>>>>>>>>>> datax-web, push process callback request, logId:{}", callback.getLogId());
  }

  public void start() {

    // valid
    if (JobExecutor.getAdminBizList() == null) {
      log.warn(">>>>>>>>>>> datax-web, executor callback config fail, adminAddresses is null.");
      return;
    }

    // callback
    processCallbackThread =
        new Thread(
            () -> {

              // normal callback
              while (!toStop) {
                try {
                  HandleProcessCallbackParam callback = getInstance().callBackQueue.take();

                  // callback list param
                  List<HandleProcessCallbackParam> callbackParamList =
                      new ArrayList<HandleProcessCallbackParam>();
                  int drainToNum = getInstance().callBackQueue.drainTo(callbackParamList);
                  callbackParamList.add(callback);

                  // callback, will retry if error
                  if (!callbackParamList.isEmpty()) {
                    doCallback(callbackParamList);
                  }
                } catch (Exception e) {
                  if (!toStop) {
                    log.error(e.getMessage(), e);
                  }
                }
              }

              // last callback
              try {
                List<HandleProcessCallbackParam> callbackParamList =
                    new ArrayList<HandleProcessCallbackParam>();
                int drainToNum = getInstance().callBackQueue.drainTo(callbackParamList);
                if (CollUtil.isNotEmpty(callbackParamList)) {
                  doCallback(callbackParamList);
                }
              } catch (Exception e) {
                if (!toStop) {
                  log.error(e.getMessage(), e);
                }
              }
              log.info(">>>>>>>>>>> datax-web, executor callback thread destory.");
            });
    processCallbackThread.setDaemon(true);
    processCallbackThread.setName("datax-web, executor TriggerCallbackThread");
    processCallbackThread.start();

    // retry
    processRetryCallbackThread =
        new Thread(
            () -> {
              while (!toStop) {
                try {
                  retryFailCallbackFile();
                } catch (Exception e) {
                  if (!toStop) {
                    log.error(e.getMessage(), e);
                  }
                }
                try {
                  TimeUnit.SECONDS.sleep(RegistryConfig.BEAT_TIMEOUT);
                } catch (InterruptedException e) {
                  if (!toStop) {
                    log.error(e.getMessage(), e);
                  }
                }
              }
              log.info(">>>>>>>>>>> datax-web, executor retry callback thread destory.");
            });
    processRetryCallbackThread.setDaemon(true);
    processRetryCallbackThread.start();
  }

  public void toStop() {
    toStop = true;
    // stop callback, interrupt and wait
    if (processCallbackThread != null) { // support empty admin address
      processCallbackThread.interrupt();
      try {
        processCallbackThread.join();
      } catch (InterruptedException e) {
        log.error(e.getMessage(), e);
      }
    }

    // stop retry, interrupt and wait
    if (processRetryCallbackThread != null) {
      processRetryCallbackThread.interrupt();
      try {
        processRetryCallbackThread.join();
      } catch (InterruptedException e) {
        log.error(e.getMessage(), e);
      }
    }
  }

  // ---------------------- fail-callback file ----------------------

  /**
   * do callback, will retry if error
   *
   * @param callbackParamList
   */
  private void doCallback(List<HandleProcessCallbackParam> callbackParamList) {
    boolean callbackRet = false;
    // callback, will retry if error
    for (AdminBiz adminBiz : JobExecutor.getAdminBizList()) {
      try {
        ReturnT<String> callbackResult = adminBiz.processCallback(callbackParamList);
        if (callbackResult != null && ReturnT.SUCCESS_CODE == callbackResult.getCode()) {
          callbackLog(callbackParamList, "<br>----------- datax-web job callback finish.");
          callbackRet = true;
          break;
        } else {
          callbackLog(
              callbackParamList,
              "<br>----------- datax-web job callback fail, callbackResult:" + callbackResult);
        }
      } catch (Exception e) {
        callbackLog(
            callbackParamList,
            "<br>----------- datax-web job callback error, errorMsg:" + e.getMessage());
      }
    }
    if (!callbackRet) {
      appendFailCallbackFile(callbackParamList);
    }
  }

  /** callback log */
  private void callbackLog(List<HandleProcessCallbackParam> callbackParamList, String logContent) {
    for (HandleProcessCallbackParam callbackParam : callbackParamList) {
      String logFileName =
          JobFileAppender.makeLogFileName(
              new Date(callbackParam.getLogDateTime()), callbackParam.getLogId());
      JobFileAppender.CONTEXT_HOLDER.set(logFileName);
      JobLogger.log(logContent);
    }
  }

  private void appendFailCallbackFile(
      List<HandleProcessCallbackParam> handleProcessCallbackParams) {
    // valid
    if (CollUtil.isEmpty(handleProcessCallbackParams)) {
      return;
    }

    // append file
    byte[] callbackParamList_bytes =
        JobExecutor.getSerializer().serialize(handleProcessCallbackParams);

    File callbackLogFile =
        new File(failCallbackFileName.replace("{x}", String.valueOf(System.currentTimeMillis())));
    if (callbackLogFile.exists()) {
      for (int i = 0; i < 100; i++) {
        callbackLogFile =
            new File(
                failCallbackFileName.replace(
                    "{x}",
                    String.valueOf(System.currentTimeMillis())
                        .concat("-")
                        .concat(String.valueOf(i))));
        if (!callbackLogFile.exists()) {
          break;
        }
      }
    }
    FileUtil.writeFileContent(callbackLogFile, callbackParamList_bytes);
  }

  private void retryFailCallbackFile() {

    // valid
    File callbackLogPath = new File(failCallbackFilePath);
    if (!callbackLogPath.exists()) {
      return;
    }
    if (callbackLogPath.isFile()) {
      callbackLogPath.delete();
    }
    if (!(callbackLogPath.isDirectory() && ArrayUtil.isNotEmpty(callbackLogPath.list()))) {
      return;
    }

    // load and clear file, retry
    List<HandleProcessCallbackParam> params;
    for (File f : callbackLogPath.listFiles()) {
      byte[] ps = FileUtil.readFileContent(f);
      params =
          (List<HandleProcessCallbackParam>)
              JobExecutor.getSerializer().deserialize(ps, HandleProcessCallbackParam.class);
      f.delete();
      doCallback(params);
    }
  }
}
